package com.yhou.cananl.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author houyong
 * @ProjectName search
 * @Description:  单线程版（创建线程池，监控到后扔到线程池中进行处理）
 * @date 2018/09/05 下午5:54
 */
@Slf4j
@Component
public class DemoCanalClient extends AbstractCanalClient {
    private static Logger log = LoggerFactory.getLogger(DemoCanalClient.class);


    @Value("${canal.business.demoDestination}")
    private String destination;

    ExecutorService canalItemConsumser;


    @PostConstruct
    public void connectCanalSubscribeItem() {
        log.info("监控商品destination={}", destination);
        if (StringUtils.isNotEmpty(destination)) {
            String[] itemDestinationArr = destination.split(",");
            canalItemConsumser = Executors.newFixedThreadPool(itemDestinationArr.length, new NamedThreadFactory("canal-demo-consumer"));
            for (String itemDestination : itemDestinationArr) {
                ConsumerRunnable itemConsumer = new ConsumerRunnable(itemDestination);
                canalItemConsumser.submit(itemConsumer);
            }
        }
    }

    @Override
    boolean insertColumn(List<CanalEntry.Column> columns, CanalEntry.Entry entry) {
        log.info("创建索引。。。。。。");
        return true;
    }

    @Override
    boolean updateColumn(List<CanalEntry.Column> columns, CanalEntry.Entry entry) {
        log.info("更新索引。。。。。。");
        return true;
    }

    @Override
    boolean deleteColumn(List<CanalEntry.Column> columns, CanalEntry.Entry entry) {
        log.info("删除索引。。。。。。");
        return true;
    }
    class ConsumerRunnable implements Runnable {
        CanalConnector connector = null;
        String itemDestination;

        public ConsumerRunnable(String itemDestination) {
            this.itemDestination = itemDestination;
//            connector = CanalConnectors.newClusterConnector(zk, itemDestination, username, password);
            connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalhost,
                    canalport), destination, "", "");
        }

        @Override
        public void run() {
            long batchId = 0;
            while (true) {
                try {
                    connector.connect();
                    //订阅那些表
                    //1.  所有表：.*   or  .*\\..*
                    //2.  canal schema下所有表： canal\\..*
                    //3.  canal下的以canal打头的表：canal\\.canal.*
                    //4.  canal schema下的一张表：canal.test1
                    //5.  多个规则组合使用：canal\\..*,mysql.test1,mysql.test2
                    //订阅那些表
//                    connector.subscribe("demo\\.*");
                    while (true) {
                        Message messages = connector.getWithoutAck(100);
                        batchId = messages.getId();
                        int size = messages.getEntries().size();
                        if (batchId == -1 || size == 0) {
                            try {
                                Thread.sleep(1000);
                                connector.ack(batchId);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            log.info("{} = No DATA change!batchId={}", itemDestination, batchId);
                        } else {
                            handleCanalEntry(messages.getEntries());
                            connector.ack(batchId);
                            log.info(" consume canal  {}   is  success!!! ", batchId);
                        }
                    }
                } catch (Exception e) {
                    // 处理失败, 按偏移量回滚数据
                    connector.rollback(batchId);
                    log.error("subscribe canal  itemDestination {} is crash!!! batchId={}", itemDestination, batchId, e);

                } finally {
//                connector.disconnect();
                }
            }
        }
    }
}
